package com.xixi.canal.client.starter.loader;

import com.alibaba.otter.canal.instance.core.CanalInstance;
import com.alibaba.otter.canal.instance.core.CanalInstanceGenerator;
import com.alibaba.otter.canal.instance.manager.CanalInstanceWithManager;
import com.alibaba.otter.canal.instance.manager.model.Canal;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter;
import com.alibaba.otter.canal.parse.inbound.mysql.AbstractMysqlEventParser;
import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DefaultTableMetaTSDBFactory;
import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDB;
import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDBBuilder;
import com.xixi.canal.client.starter.config.CanalConfig;
import com.xixi.canal.client.starter.config.InstanceConfig;
import com.xixi.canal.client.support.MasterInfo;
import org.apache.commons.lang.BooleanUtils;

import java.net.InetSocketAddress;
import java.util.Random;

public class ClientCanalInstanceGenerator implements CanalInstanceGenerator {

    private final InstanceConfig instanceConfig;
    private final CanalConfig canalConfig;

    public ClientCanalInstanceGenerator(CanalConfig canalConfig, InstanceConfig instanceConfig) {
        this.instanceConfig = instanceConfig;
        this.canalConfig = canalConfig;
    }

    @Override
    public CanalInstance generate(String destination) {
        Canal canal = buildCanal(destination);

        CanalInstanceWithManager canalInstanceWithManager = new CanalInstanceWithManager(canal, instanceConfig.getFilter());

        boolean tsdbEnable = BooleanUtils.toBoolean(instanceConfig.getEnableTsdb());
        if (tsdbEnable) {
            AbstractMysqlEventParser eventParser = (AbstractMysqlEventParser)canalInstanceWithManager.getEventParser();

            eventParser.setTsdbSpringXml(instanceConfig.getTsdbSpringXml());

            eventParser.setTableMetaTSDBFactory(new DefaultTableMetaTSDBFactory(){
                @Override
                public void destory(String destination) {
                    TableMetaTSDBBuilder.destory(destination);
                }

                @Override
                public TableMetaTSDB build(String destination, String springXml) {
                    try {
                        CanalParameter parameters = canal.getCanalParameter();
                        System.setProperty("canal.instance.tsdb.url", parameters.getTsdbJdbcUrl());
                        System.setProperty("canal.instance.tsdb.dbUsername", parameters.getTsdbJdbcUserName());
                        System.setProperty("canal.instance.tsdb.dbPassword", parameters.getTsdbJdbcPassword());

                        return TableMetaTSDBBuilder.build(destination, springXml);
                    } finally {
                        System.setProperty("canal.instance.tsdb.url", "");
                        System.setProperty("canal.instance.tsdb.dbUsername", "");
                        System.setProperty("canal.instance.tsdb.dbPassword", "");
                    }
                }
            });
            eventParser.setEnableTsdb(tsdbEnable);
        }
        return canalInstanceWithManager;
    }

    private Canal buildCanal(String destination) {
        Canal canal = new Canal();
        Random random = new Random(100000);
        canal.setId(random.nextLong());
        canal.setName(destination);
        canal.setDesc(destination);


        CanalParameter parameter = new CanalParameter();

        // META
        // MEMORY/MIXED/LOCAL_FILE
        initMeta(parameter);

        parameter.setSourcingType(CanalParameter.SourcingType.MYSQL);
        initMasterInfo(parameter);

        // Event Store
        initStore(parameter);

        parameter.setHaMode(CanalParameter.HAMode.HEARTBEAT);

        initParser(parameter);

        canal.setCanalParameter(parameter);


        return canal;
    }

    private void initParser(CanalParameter parameter) {
        parameter.setDetectingEnable(instanceConfig.getDetectingEnable());
        parameter.setDetectingIntervalInSeconds(instanceConfig.getDetectingIntervalTime());
        parameter.setDetectingRetryTimes(instanceConfig.getDetectingRetryThreshold());
        parameter.setDetectingSQL(instanceConfig.getDetectingSql());

        parameter.setSlaveId(instanceConfig.getMaster().getSlaveId());

//        网络链接参数
        parameter.setDefaultConnectionTimeoutInSeconds(instanceConfig.getNetworkSoTimeout());
        parameter.setConnectionCharset(instanceConfig.getMaster().getConnectionCharset());
        parameter.setReceiveBufferSize(instanceConfig.getNetworkReceiveBufferSize());
        parameter.setSendBufferSize(instanceConfig.getNetworkSendBufferSize());

        parameter.setFallbackIntervalInSeconds(instanceConfig.getFallbackIntervalInSeconds());

        parameter.setParallel(instanceConfig.getParallel());
        parameter.setTransactionSize(instanceConfig.getTransactionSize());

//        parameter.setTsdbEnable(instanceConfig.getEnableTsdb());
        parameter.setTsdbEnable(false);
        parameter.setTsdbJdbcUserName(instanceConfig.getTsdbDbUsername());
        parameter.setTsdbJdbcPassword(instanceConfig.getTsdbDbPassword());
        parameter.setTsdbJdbcUrl(instanceConfig.getTsdbUrl());

        String indexMode = instanceConfig.getIndexMode();
        // MEMORY/ZOOKEEPER/MIXED/META/MEMORY_META_FAILBACK
        if ("MEMORY".equalsIgnoreCase(indexMode)) {
            parameter.setIndexMode(CanalParameter.IndexMode.MEMORY);
        } else if ("ZOOKEEPER".equalsIgnoreCase(indexMode)) {
            parameter.setIndexMode(CanalParameter.IndexMode.ZOOKEEPER);
        } else if ("MIXED".equalsIgnoreCase(indexMode)) {
            parameter.setIndexMode(CanalParameter.IndexMode.MIXED);
        } else if ("META".equalsIgnoreCase(indexMode)) {
            parameter.setIndexMode(CanalParameter.IndexMode.META);
        } else if ("MEMORY_META_FAILBACK".equalsIgnoreCase(indexMode)) {
            parameter.setIndexMode(CanalParameter.IndexMode.MEMORY_META_FAILBACK);
        } else {
            throw new IllegalArgumentException("Index mode not support.");
        }
    }

    private void initMasterInfo(CanalParameter parameter) {

        MasterInfo master = instanceConfig.getMaster();
        parameter.setMasterAddress(new InetSocketAddress(master.getAddress(), master.getPort()));
        parameter.setMasterUsername(master.getUserName());
        parameter.setMasterPassword(master.getPassword());
        parameter.setMasterLogfileName(master.getJournalName());
        parameter.setMasterLogfileOffest(master.getPosition());
        parameter.setMasterTimestamp(master.getTimestamp());

    }

    private void initStore(CanalParameter parameter) {
        parameter.setStorageMode(CanalParameter.StorageMode.MEMORY);
        parameter.setMemoryStorageBufferSize(instanceConfig.getMemoryBufferSize());
        parameter.setMemoryStorageBufferMemUnit(instanceConfig.getMemoryBufferMemunit());
        parameter.setMemoryStorageRawEntry(instanceConfig.getMemoryRawEntry());
        parameter.setDdlIsolation(instanceConfig.getDdlIsolation());
    }

    private void initMeta(CanalParameter parameter) {
        String metaMode = canalConfig.getMetaMode();
        if ("FILE".equalsIgnoreCase(metaMode)) {
            parameter.setMetaMode(CanalParameter.MetaMode.LOCAL_FILE);
            parameter.setDataDir(canalConfig.getFileDataDir());
            parameter.setMetaFileFlushPeriod(canalConfig.getFileFlushPeriod());
        } else if("MEMORY".equalsIgnoreCase(metaMode)) {
            parameter.setMetaMode(CanalParameter.MetaMode.MEMORY);
        } else if ("MIXED".equalsIgnoreCase(metaMode)) {
            parameter.setMetaMode(CanalParameter.MetaMode.MIXED);
        } else {
            throw new IllegalArgumentException("Meta Mode " +
                    metaMode + "cannot support.");
        }
    }
}
